package checkpoint;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.runtime.state.storage.FileSystemCheckpointStorage;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import javax.ws.rs.DELETE;
import java.time.Duration;

/**
 * @date 2024/11/22 17:09
 *
 * 检查点原理理解，会组合起来使用即可
 * 检查点重置  学习怎么使用
 */
//todo 未测
public class Flink01_CheckPointConfig {
    public static void main(String[] args) {
        //设置HDFS的用户,用于访问HDFS，存储检查点
        System.setProperty("HADOOP_USER_NAME" ,"atguigu");

        //设置网页地址
        Configuration conf = new Configuration();
        conf.setInteger("rest.port",5678);

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);

        //设置状态后端,默认就是HashMapStateBackend
        env.setStateBackend(new HashMapStateBackend());

        //开启检查点，5秒保存一次，默认是精准一次,怎么查看，不会
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);

        //检查点配置,获得检查点配置对象
        CheckpointConfig checkpointConfig = env.getCheckpointConfig();

        //检查点存储位置：两个位置
          //JobManagerCheckpointStorage,默认情况下将检查点存储到JobManager的内存中
          //FileSystemCheckpointStorage,将减产点存储到指定的文件系统中，例如HDFS
          //todo 实际开发使用，1、需要加hadoop-client的依赖。2、设置hdfs登录用户，用于访问hdfs
        checkpointConfig.setCheckpointStorage(
                new FileSystemCheckpointStorage("hdfs://hadoop102:8020/flink-ck")
        );

         //检查点的间隔，和上面有啥区别
        //和上面没区别，只是在配置和上面同时设置优先执行配置中的
        checkpointConfig.setCheckpointInterval(3000l); //以配置为准
        //和上面没区别，只是在配置和上面同时设置优先执行配置中的
        //检查点模式
        checkpointConfig.setCheckpointingMode(CheckpointingMode.AT_LEAST_ONCE);

        //检查点超时时间，超时没完成就会被丢弃掉。传入一个长整型毫秒数作为参数，表示超时时间。
        checkpointConfig.setCheckpointTimeout(10000l);//设置这个可以用来干嘛呀？10s

        //同时执行的检查点个数
        checkpointConfig.setMaxConcurrentCheckpoints(1);//可调吗？

        /** //两个检查点之间的最小间隔
        checkpointConfig.setMinPauseBetweenCheckpoints(3000L);//时间之间的关系

        //作业取消后，是否保留外部检查点
        //CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION; //保留，这里为什么会报红呢
        //CheckpointConfig.ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION   //删除
        checkpointConfig.setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION);

        //检查点最大失败的个数
        checkpointConfig.setTolerableCheckpointFailureNumber(3);

        //允许开始非对齐
        checkpointConfig.enableUnalignedCheckpoints();

        //Barrier对齐的超时时间
        checkpointConfig.setAlignedCheckpointTimeout(Duration.ofSeconds(5));

        //强制开启非对齐
        checkpointConfig.setForceUnalignedCheckpoints(true);*/

        env.socketTextStream("hadoop102",8888).print();//不写会报错



        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

}
